package com.example.emos.wx.task;

import com.example.emos.wx.db.pojo.MessageEntity;
import com.example.emos.wx.db.pojo.MessageRefEntity;
import com.example.emos.wx.exception.EmosException;
import com.example.emos.wx.service.MessageService;
import com.rabbitmq.client.*;
import lombok.extern.slf4j.Slf4j;
import org.springframework.beans.factory.annotation.Autowired;
import org.springframework.scheduling.annotation.Async;
import org.springframework.stereotype.Component;

import java.util.HashMap;
import java.util.Map;

@Component
@Slf4j
public class MessageTask {
    @Autowired
    private ConnectionFactory factory;

    @Autowired
    private MessageService messageService;

    /**
     * 向rabbitmq发送消息（同步）
     * @param topic:消息的名字  主题
     * @param entity 消息对象
     */
    public void send(String topic, MessageEntity entity) {
        // 向MongoDB保存消息数据，返回消息ID
        String id = messageService.insertMessage(entity);

        //向RabbitMQ发送消息
        // 创建连接：Connection connection = factory.newConnection()
        // 创建通道：Channel channel = connection.createChannel()
        try(Connection connection = factory.newConnection();
            Channel channel = connection.createChannel();)
        {
            // 连接到某个Topic
            // topic:连接的topic队列(不存在就创建) true:是否持久化 false:是否排他(加锁)
            // false:是否自动删除队列 null；hashMap对象
            channel.queueDeclare(topic, true, false, false, null);
            HashMap map = new HashMap();
            map.put("messageId", id);
            AMQP.BasicProperties properties = new AMQP.BasicProperties().builder().headers(map).build();
            channel.basicPublish("", topic, properties, entity.getMsg().getBytes());
            log.debug("消息发送成功");
        } catch (Exception e) {
            log.error("执行异常", e);
            throw new EmosException("向MQ发送消息失败");
        }
    }

    @Async
    public void sendAsync(String topic, MessageEntity entity) {
        send(topic, entity);
    }

    public int receive(String topic) {
        int i = 0; // 累加消息接收数量
        try(Connection connection = factory.newConnection();
            Channel channel = connection.createChannel();)
        {
            channel.queueDeclare(topic, true, false, false, null);
            while(true) {
                // false:接收到消息是否自动返回ack应答,这里返回false是因为还得把数据写入集合中
                // 写入集合没问题的话就手动返回
                GetResponse response = channel.basicGet(topic, false);
                if(response != null) {
                    AMQP.BasicProperties properties = response.getProps(); // 从响应里拿到绑定的数据
                    Map<String, Object> map = properties.getHeaders();
                    String messageId = map.get("messageId").toString();
                    byte[] body = response.getBody();
                    String message = new String(body);
                    log.debug("从RabbitMQ接收的消息：" + message);

                    MessageRefEntity entity = new MessageRefEntity();
                    entity.setMessageId(messageId);
                    entity.setReceiverId(Integer.parseInt(topic));
                    entity.setReadFlag(false);
                    entity.setLastFlag(true);
                    messageService.inserRef(entity);
                    long deliveryTag = response.getEnvelope().getDeliveryTag();
                    channel.basicAck(deliveryTag, false); // 返回ack应答
                    i++;
                } else {
                    break;
                }
            }
        } catch (Exception e) {
            log.error("执行异常", e);
            throw new EmosException("接收消息失败");
        }
        return i;
    }

    @Async
    public int receiveAsync(String topic) {
        return receive(topic);
    }

    // 删除消息队列
    public void deleteQueue(String topic) {
        try(Connection connection = factory.newConnection();
            Channel channel = connection.createChannel();)
        {
            channel.queueDelete(topic);
            log.debug("消息队列成功删除");
        } catch (Exception e) {
            log.error("删除队列失败", e);
            throw new EmosException("删除队列失败");
        }
    }

    @Async
    public void deleteQueueAsync(String topic) {
        deleteQueue(topic);
    }

}
